package drds.binlog;

import drds.binlog.event.FormatDescriptionLogEvent;

import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.Arrays;

/**
 * TODO: Document It!!
 * <p>
 * <pre>
 * FileLogFetcher fetcher = new FileLogFetcher();
 * fetcher.open(file, 0);
 *
 * while (fetcher.fetch()) {
 *     LogEvent event;
 *     do {
 *         event = decoder.decode(fetcher, context);
 *
 *         // process log event.
 *     } while (event != null);
 * }
 * // file ending reached.
 * </pre>
 */
public final class FileLogFetcher extends LogFetcher
{

    public static final byte[] BINLOG_MAGIC = {-2, 0x62, 0x69, 0x6e};

    private FileInputStream fileInputStream;

    public FileLogFetcher()
    {
        super(default_initial_capacity, default_growth_factor);
    }

    public FileLogFetcher(final int initialCapacity)
    {
        super(initialCapacity, default_growth_factor);
    }

    public FileLogFetcher(final int initialCapacity, final float growthFactor)
    {
        super(initialCapacity, growthFactor);
    }

    /**
     * Open binlog file in local disk to fetch.
     */
    public void open(File file) throws FileNotFoundException, IOException
    {
        open(file, 0L);
    }

    /**
     * Open binlog file in local disk to fetch.
     */
    public void open(String filePath) throws FileNotFoundException, IOException
    {
        open(new File(filePath), 0L);
    }

    /**
     * Open binlog file in local disk to fetch.
     */
    public void open(String filePath, final long filePosition) throws FileNotFoundException, IOException
    {
        open(new File(filePath), filePosition);
    }

    /**
     * Open binlog file in local disk to fetch.
     */
    public void open(File file, final long filePosition) throws FileNotFoundException, IOException
    {
        fileInputStream = new FileInputStream(file);

        ensureCapacity(bin_log_header_size);
        if (bin_log_header_size != fileInputStream.read(bytes, 0, bin_log_header_size))
        {
            throw new IOException("No binlog file headerPacket");
        }

        if (bytes[0] != BINLOG_MAGIC[0] || bytes[1] != BINLOG_MAGIC[1] || bytes[2] != BINLOG_MAGIC[2]
                || bytes[3] != BINLOG_MAGIC[3])
        {
            throw new IOException("Error binlog file headerPacket: "
                    + Arrays.toString(Arrays.copyOf(bytes, bin_log_header_size)));
        }

        limit = 0;
        originIndex = 0;
        index = 0;

        if (filePosition > bin_log_header_size)
        {
            final int maxFormatDescriptionEventLen = FormatDescriptionLogEvent.LOG_EVENT_MINIMAL_HEADER_LEN
                    + FormatDescriptionLogEvent.ST_COMMON_HEADER_LEN_OFFSET
                    + LogEvent.ENUM_END_EVENT + LogEvent.BINLOG_CHECKSUM_ALG_DESC_LEN
                    + LogEvent.CHECKSUM_CRC32_SIGNATURE_LEN;

            ensureCapacity(maxFormatDescriptionEventLen);
            limit = fileInputStream.read(bytes, 0, maxFormatDescriptionEventLen);
            limit = (int) getUint32(LogEvent.EVENT_LEN_OFFSET);
            fileInputStream.getChannel().position(filePosition);
        }
    }

    /**
     * {@inheritDoc}
     *
     * @see LogFetcher#fetch()
     */
    public boolean fetch() throws IOException
    {
        if (limit == 0)
        {
            final int length = fileInputStream.read(bytes, 0, bytes.length);
            if (length >= 0)
            {
                limit += length;
                index = 0;
                originIndex = 0;

                /* More binlog to fetch */
                return true;
            }
        } else if (originIndex == 0)
        {
            if (limit > bytes.length / 2)
            {
                ensureCapacity(bytes.length + limit);
            }
            final int len = fileInputStream.read(bytes, limit, bytes.length - limit);
            if (len >= 0)
            {
                limit += len;

                /* More binlog to fetch */
                return true;
            }
        } else if (limit > 0)
        {
            if (limit >= FormatDescriptionLogEvent.LOG_EVENT_HEADER_LEN)
            {
                int lenPosition = index + 4 + 1 + 4;
                long eventLen = ((long) (0xff & bytes[lenPosition++])) | ((long) (0xff & bytes[lenPosition++]) << 8)
                        | ((long) (0xff & bytes[lenPosition++]) << 16)
                        | ((long) (0xff & bytes[lenPosition++]) << 24);

                if (limit >= eventLen)
                {
                    return true;
                } else
                {
                    ensureCapacity((int) eventLen);
                }
            }

            System.arraycopy(bytes, originIndex, bytes, 0, limit);
            index -= originIndex;
            originIndex = 0;
            final int len = fileInputStream.read(bytes, limit, bytes.length - limit);
            if (len >= 0)
            {
                limit += len;

                /* More binlog to fetch */
                return true;
            }
        } else
        {
            /* Should not happen. */
            throw new IllegalArgumentException("Unexcepted limit: " + limit);
        }

        /* Reach binlog file end */
        return false;
    }

    /**
     * {@inheritDoc}
     *
     * @see LogFetcher#close()
     */
    public void close() throws IOException
    {
        if (fileInputStream != null)
        {
            fileInputStream.close();
        }

        fileInputStream = null;
    }
}
